package com.xy.tool.data.redis.ex;

import com.xy.lang.util.CollectionUtilsEx;
import com.xy.tool.data.redis.util.RedisUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;

import static com.xy.tool.data.redis.ex.RedisTemplateEx.DEFAULT_MULTI_OPERATION_BATCH_SIZE;
import static com.xy.tool.data.redis.util.TimeUtils.getExpiredSecondsVariation;

public class ValueOperationsEx<V> {

    private final RedisTemplate<String, V> redisTemplate;
    private final ValueOperations<String, V> valueOps;

    public ValueOperationsEx(RedisTemplate<String, V> redisTemplate) {
        Objects.requireNonNull(redisTemplate);
        this.redisTemplate = redisTemplate;
        this.valueOps = redisTemplate.opsForValue();
    }

    public final void set(String key, V value, long seconds) {
        if(seconds < 1) {
            valueOps.set(key, value);
        } else {
            valueOps.set(key, value,
                    seconds + getExpiredSecondsVariation(seconds),
                    TimeUnit.SECONDS);
        }
    }

    public final void multiSet(Map<String, V> map, long seconds) {
        multiSet(map, seconds, DEFAULT_MULTI_OPERATION_BATCH_SIZE);
    }

    public final void multiSet(Map<String, V> map, long seconds, int batchSize) {
        if (CollectionUtils.isEmpty(map)) {
            return;
        }
        if (batchSize < 1) {
            batchSize = DEFAULT_MULTI_OPERATION_BATCH_SIZE;
        }
        if(map.size() > batchSize) {
            CollectionUtilsEx.foreachSubMap(map, batchSize,
                    (loop, startIndex, endIndex, subMap) -> {
                        multiSet0(subMap, seconds);
                    });
        } else {
            multiSet0(map, seconds);
        }
    }

    private void multiSet0(Map<String, V> map, long seconds) {
        valueOps.multiSet(map);
        if(seconds > 0) {
            RedisUtils.expire(redisTemplate, map.keySet(), seconds);
        }
    }

    public final List<V> multiGet(Collection<String> keys) {
        return multiGet(keys, DEFAULT_MULTI_OPERATION_BATCH_SIZE);
    }

    public final List<V> multiGet(Collection<String> keys, int batchSize) {
        if (CollectionUtils.isEmpty(keys)) {
            return Collections.emptyList();
        }
        if (batchSize < 1) {
            batchSize = DEFAULT_MULTI_OPERATION_BATCH_SIZE;
        }
        if(keys.size() > batchSize) {
            List<V> list = new ArrayList<>();
            CollectionUtilsEx.foreachSubList(new ArrayList<>(keys), batchSize,
                    (loop, startIndex, endIndex, subList) -> {
                        list.addAll(valueOps.multiGet(subList));
                    });
            return list;
        }
        return valueOps.multiGet(keys);
    }

    public final <R> Map<String, R> increment(Collection<String> keys, R delta) {
        return increment(keys, delta, DEFAULT_MULTI_OPERATION_BATCH_SIZE);
    }

    public final <R> Map<String, R> increment(Collection<String> keys, R delta, int batchSize) {
        if (CollectionUtils.isEmpty(keys)) {
            return Collections.emptyMap();
        }
        if (batchSize < 1) {
            batchSize = DEFAULT_MULTI_OPERATION_BATCH_SIZE;
        }
        if(keys.size() > batchSize) {
            Map<String, R> valueMap = new HashMap<>();
            CollectionUtilsEx.foreachSubList(new ArrayList<>(keys), batchSize,
                    (loop, startIndex, endIndex, subList) -> {
                        valueMap.putAll(increment0(subList, delta));
                    });
            return valueMap;
        }
        return increment0(keys, delta);
    }

    private <R> Map<String, R> increment0(Collection<String> keys, R delta) {
        List<Object> valueList = redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Nullable
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                if (delta instanceof Long) {
                    for (String key : keys) {
                        connection.incrBy(RedisSerializer.string().serialize(key), (Long) delta);
                    }
                } else {
                    for (String key : keys) {
                        connection.incrBy(RedisSerializer.string().serialize(key), (Double) delta);
                    }
                }
                return null;
            }
        });
        Map<String, R> valueMap = new HashMap<>();
        Iterator<String> it = keys.iterator();
        for(int i = 0, len = keys.size(); i < len; i++) {
            valueMap.put(it.next(), (R)valueList.get(i));
        }
        return valueMap;
    }

}
